package com.uouie.common;

import com.uouie.common.config.IMqttDataValidator;
import com.uouie.common.entity.MqttData;
import com.uouie.common.service.IMqttService;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Administrator
 */
@Service
public class MqttConnect {
    /**
     * 连接地址
     */
    private String host;
    /**
     * 连接clientId，不可相同
     */
    private String clientId;
    /**
     * 登录账户
     */
    private String username;
    /**
     * 登录密码
     */
    private String password;
    /**
     * 订阅标题
     */
    private String subTopic;
    /**
     * 默认qos等级
     */
    private Integer qos;
    /**
     * 连接超时时间
     */
    private Integer completionTimeout;
    /**
     * 心跳包发送时间
     */
    private Integer keepAlive;
    /**
     * 连接配置
     */
    private MqttConnectOptions options;
    /**
     * MemoryPersistence设置clientId的保存方式
     */
    private MemoryPersistence persistence = null;
    /**
     * 异步MQTT客户端
     */
    private MqttAsyncClient mqttClient = null;
    /**
     * 发布的初始订阅
     */
    private String pubTopic;
    /**
     * 将 Spring 中所有实现 IMqttService 的接口类注入到这个Map中
     */
    private final Map<String, IMqttService> mqttImplMap = new ConcurrentHashMap<>();
    /**
     * 实现这个接口就能自定义接收数据的验证器，默认全部通过
     */
    @Autowired(required = false)
    IMqttDataValidator mqttDataValidator;

    public MqttConnect(@Value("${mqtt.host:tcp://localhost:1883}") String host,
                       @Value("${mqtt.clientId:65444}") String clientId,
                       @Value("${mqtt.username:guest}") String username,
                       @Value("${mqtt.password:guest}") String password,
                       @Value("${mqtt.subTopic:subTopic.#}") String subTopic,
                       @Value("${mqtt.qos:1}") Integer qos,
                       @Value("${mqtt.completionTimeout:3000}") Integer completionTimeout,
                       @Value("${mqtt.keepAlive:20}") Integer keepAlive,
                       @Value("${mqtt.pubTopic:pubTopic}") String pubTopic,
                       Map<String, IMqttService> strategyMap) throws MqttException {
        this.host = host;
        this.clientId = clientId;
        this.username = username;
        this.password = password;
        this.subTopic = subTopic;
        this.qos = qos;
        this.completionTimeout = completionTimeout;
        this.keepAlive = keepAlive;
        this.mqttImplMap.clear();
        strategyMap.forEach((k, v) -> this.mqttImplMap.put(v.getVersion(), v));
        if (mqttDataValidator == null) {
            this.mqttDataValidator = msg -> true;
        }
        //默认执行构造函数后执行初始化，自动连接MQTT服务器
        connect();
    }

    private IMqttService getMqttService(String serviceName) {
        IMqttService mqttService = mqttImplMap.get(serviceName);
        if (mqttService == null) {
            throw new IllegalArgumentException(serviceName + "service NOT FOUND");
        }
        return mqttService;
    }

    /**
     * 创建客户端
     */
    private void createClient() {
        if (mqttClient == null) {
            try {
              /*host为主机名，clientId是连接MQTT的客户端ID
                默认是以内存方式保存*/
                persistence = new MemoryPersistence();
                mqttClient = new MqttAsyncClient(host, clientId, persistence);
                //设置回调函数
                mqttClient.setCallback(new MqttCallbackExtended() {

                    @Override
                    public void connectComplete(boolean reconnect, String serverURI) {
                        //订阅消息
                        String[] topic = subTopic.split(",");
                        Arrays.stream(topic).forEach(item -> {
                            try {
                                mqttClient.subscribe(item, qos);
                                System.out.println("====>>>>device subscribe topic：" + item);
                            } catch (MqttException e) {
                                e.printStackTrace();
                                System.out.println("device failed to subscribe topic：" + item);
                            }
                        });
                    }

                    @Override
                    public void connectionLost(Throwable throwable) {
                        System.out.println("MQTT connect lost! try to reConnect");
                        connect();
                    }

                    //当有订阅的消息时会从这里接收
                    @Override
                    public void messageArrived(String topic, MqttMessage message) throws Exception {
                        String[] topicItem = topic.split("/");
                        String msg = new String(message.getPayload());
                        if (mqttDataValidator.check(msg)) {
                            System.out.println("MQTT Received:" + msg);
                            try {
                                //通过版本号寻找对应的服务
                                IMqttService mqttService = getMqttService(topicItem[2]);
                                //运用反射寻找对应的方法
                                Method method = mqttService.getClass().getDeclaredMethod(topicItem[3], MqttData.class);
                                // 调用目标业务 方法
                                method.invoke(mqttService, new MqttData<>());
                            } catch (IllegalArgumentException e) {
                                System.out.println(e.getMessage());
                            } catch (NoSuchMethodException | SecurityException e) { // Your exception handling goes here
                                System.out.println(topicItem[2] + "service NOT FOUND METHOD:" + topicItem[3]);
                            }
                        } else {
                            System.out.println("MQTT Format Error:" + msg);
                        }
                    }

                    @Override
                    public void deliveryComplete(IMqttDeliveryToken token) {
                        //   System.out.println("deliveryComplete---------" + token.isComplete());
                    }

                });
                System.out.println("====>>>>MQTTClient start SUCCESS");
            } catch (MqttException e) {
                System.out.println("MQTTClient connect error=" + e.getMessage());
                e.printStackTrace();
            }
        }
    }

    /**
     * 设置连接属性
     */
    private void setOptions() {
        if (options != null) {
            options = null;
        }
        options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(completionTimeout);
        options.setKeepAliveInterval(keepAlive);
        //设置自动重新连接
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        System.out.println("====>>>>set MQTT option SUCCESS");
    }

    /**
     * 连接MQTT服务器
     */
    private synchronized void connect() {
        /*设置配置*/
        if (options == null) {
            setOptions();
        }
        if (mqttClient == null) {
            createClient();
        }
        // 建立连接
        while (!mqttClient.isConnected()) {
            try {
                IMqttToken iMqttToken = mqttClient.connect(options);
                iMqttToken.waitForCompletion();
            } catch (Exception e) {
                System.out.println("=====>>>>>MQTT connect failed,message=" + e.getMessage());
                e.printStackTrace();
            }
        }
    }

    /**
     * 重新连接MQTT
     */
    public synchronized void refresh() {
        disconnect();
        setOptions();
        createClient();
        connect();
    }

    /**
     * 断开与mqtt的连接
     */
    public synchronized void disconnect() {
        //判断客户端是否null 是否连接
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                IMqttToken token = mqttClient.disconnect();
                token.waitForCompletion();
            } catch (MqttException e) {
                System.out.println("====>>>>shutdown MQTT Connect failed message={}" + e.getMessage());
            }
        }
        mqttClient = null;
    }

    /**
     * 发布
     * @param qos         连接方式
     * @param retained    是否保留
     * @param topic       主题
     * @param pushMessage 消息体
     */
    public boolean publish(int qos, boolean retained, String topic, String pushMessage) {
        System.out.println("pubTopic:" + topic);
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        try {
            IMqttDeliveryToken token = mqttClient.publish(topic, message);
            token.waitForCompletion();
            return true;
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
            return false;
        } catch (MqttException e) {
            if (e.getReasonCode() == MqttException.REASON_CODE_CLIENT_NOT_CONNECTED) {
            }
            System.out.println("=======>>>>>An error occurred while publishing the topic=" + topic + ",message=" + e.getMessage());
            return false;
        }
    }


    public void publicUtil(MqttMessage message, String version, String mac) throws MqttException {
        //没有版本号的
        if (version.isEmpty()) {
            mqttClient.publish(subTopic, message);
            //有版本号的
        } else {
            if (version.equals("v1.0.0") || version.equals("1.0.0") || version.equals("1.0.2") || version.equals("1.0.3") || version.equals("1.0.4") || version.equals("1.0.5") || version.equals("1.0.6") || version.equals("1.0.7")) {
                mqttClient.publish(subTopic, message);
            } else {
                String wildcard = subTopic + "/" + mac;
                System.out.println("主题：" + wildcard);
                mqttClient.publish(wildcard, message);
            }
        }
    }

    public void publishByte(byte[] bytes, String version, String mac) throws MqttException {
        // 创建消息
        MqttMessage message = new MqttMessage(bytes);
        // 设置消息的服务质量
        message.setQos(qos);
        // 发布消息
        if (!mqttClient.isConnected()) {
            reConnectAndPublish(message, version, mac);
        } else {
            publicUtil(message, version, mac);
        }

    }

    public synchronized void reConnectAndPublish(MqttMessage message, String version, String mac) throws MqttException {
        connect();
        new Thread(() -> {
            try {
                Thread.sleep(500);
                publicUtil(message, version, mac);
            } catch (InterruptedException e) {
                e.printStackTrace();
                System.out.println(e.getMessage());
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
                System.out.println(e.getMessage());
            } catch (MqttException e) {
                e.printStackTrace();
                System.out.println(e.getMessage());
            }
        }).start();
    }

    public Integer getQos() {
        return qos;
    }

    public String getPubTopic() {
        return pubTopic;
    }

    /**
     * 是否处于连接状态
     */
    public boolean isConnected() {
        return mqttClient != null && mqttClient.isConnected();
    }

    public String getClientId() {
        return clientId;
    }

    ;
}